package com.gandong8.tablestore.tool.core;


import com.alicloud.openservices.tablestore.model.StreamRecord;
import com.alicloud.openservices.tablestore.tunnel.worker.IChannelProcessor;
import com.alicloud.openservices.tablestore.tunnel.worker.ProcessRecordsInput;
import com.gandong8.tablestore.tool.service.inter.TablestoreService;
import com.gandong8.tablestore.tool.util.TableStoreUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.List;


@Slf4j
public class TsProcessor implements IChannelProcessor {
    private TablestoreService tablestoreService;
    private Class<?> sourceTable;
    private String tunnelId;


    TsProcessor(TablestoreService tablestoreService, String tunnelId, Class<?> sourceTable) {
        this.tablestoreService = tablestoreService;
        this.tunnelId = tunnelId;
        this.sourceTable = sourceTable;
    }

    @Override
    public void process(ProcessRecordsInput input) {
        try {
            dealData(input.getRecords());
        } catch (Exception e) {
            log.error("processException", e);
        }
    }

    private void dealData(List<StreamRecord> recordList){
        if(recordList.size() < 1) {
            return;
        }
        for(StreamRecord record:recordList) {
            //PUT, UPDATE， DELETE
            String recordType = record.getRecordType().name();
            if("DELETE".equals(recordType)) {
                //todo
                continue;
            }
            Object obj = TableStoreUtil.parseStreamRecord(this.sourceTable, record);
            if("PUT".equals(recordType)) {
                Object newObj = new Object();
                //todo obj->newObj
                tablestoreService.putData(newObj);
                continue;
            }
            if("UPDATE".equals(recordType)) {
                //todo
                continue;
            }
            log.error("recordTypeNotExist:{},{}", recordType, record.getPrimaryKey());
        }
    }


    @Override
    public void shutdown() {
        log.warn("通道关闭:{}", this.tunnelId);
    }
}
